package com.mini.rpc.serialization;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.mini.rpc.common.MiniRpcRequest;
import com.mini.rpc.common.MiniRpcResponse;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;

/**
 * Kryo序列化算法
 *
 * @author: hliushi
 * @date: 2022/4/23 23:08
 **/
@Slf4j
public class KryoSerialization implements RpcSerialization {

    /**
     * 因为Kryo不是线程安全的。所以，使用ThreadLocal存储Kryo对象
     */
    private final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        kryo.register(MiniRpcRequest.class);
        kryo.register(MiniRpcResponse.class);
        return kryo;
    });


    @Override
    public <T> byte[] serialize(T obj) throws IOException {
        try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
             Output output = new Output(byteArrayOutputStream)) {
            Kryo kryo = kryoThreadLocal.get();
            // Object -> byte: 将对象序列化为byte数组
            kryo.writeObject(output, obj);
            kryoThreadLocal.remove();
            return output.toBytes();
        } catch (Exception e) {
            throw new SerializationException(e);
        }
    }

    @Override
    public <T> T deserialize(byte[] data, Class<T> clz) throws IOException {
        try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(data);
             Input input = new Input(byteArrayInputStream)) {
            Kryo kryo = kryoThreadLocal.get();
            // byte -> Object: 将byte数组中反序列化出对象
            T t = kryo.readObject(input, clz);
            kryoThreadLocal.remove();
            // todo 这里的泛型使用问题?
            return clz.cast(t);
        } catch (Exception e) {
            throw new SerializationException(e);
        }
    }
}
